import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.function.Consumer;

/**
 * <p>
 *
 * </p>
 * @author wind
 * @date    2024/12/12 14:13
 * @version v1.0
 */
public class HdfsHelper {

    private static final Logger logger = LoggerFactory.getLogger(HdfsHelper.class);

    private final String hadoopConfDir;

    private final String hdfsUrl;

    private final String hdfsUser;

    private final String krb5User;

    private final String krb5Tab;

    private final String krb5Conf;

    private HdfsHelper(String hadoopConfDir, String hdfsUrl, String hdfsUser, String krb5User, String krb5Tab, String krb5Conf){
        this.hadoopConfDir = hadoopConfDir;
        this.hdfsUrl = hdfsUrl;
        this.hdfsUser = hdfsUser;
        this.krb5User = krb5User;
        this.krb5Tab = krb5Tab;
        this.krb5Conf = krb5Conf;
    }

    public static HdfsHelper fromConfig(String hadoopConfDir){
        return new HdfsHelper(hadoopConfDir, null, null, null, null, null);
    }

    public static HdfsHelper fromKrb5(String hadoopConfDir, String krb5User, String krb5Tab, String krb5Conf){
        return new HdfsHelper(hadoopConfDir, null, null, krb5User, krb5Tab, krb5Conf);
    }

    public static HdfsHelper fromUrl(String hdfsUrl, String hdfsUser){
        return new HdfsHelper(null, hdfsUrl, hdfsUser, null, null, null);
    }

    public static HdfsHelper fromUrl(String hdfsUrl){
        return new HdfsHelper(null, hdfsUrl, null, null, null, null);
    }


    public Configuration initConf(){
        Configuration conf = new Configuration();
        if(StringUtils.isNotEmpty(hadoopConfDir)){
            conf.addResource(new Path(hadoopConfDir + "/core-site.xml"));
            conf.addResource(new Path(hadoopConfDir + "/hdfs-site.xml"));
        }
        if(isKerberos()){
            System.setProperty("java.security.krb5.conf", krb5Conf);
            try {
                UserGroupInformation.setConfiguration(conf);
                UserGroupInformation.loginUserFromKeytab(krb5User, krb5Tab);
            } catch (IOException e) {
                logger.error(e.getMessage());
            }
        }
        return conf;
    }

    /**
     *
     * @return
     * @throws URISyntaxException
     * @throws IOException
     * @throws InterruptedException
     */
    public FileSystem getFileSystem() throws Exception {
        Configuration conf = initConf();
        if(isConfig()){
            return FileSystem.get(conf);
        }
        URI u = new URI(hdfsUrl);
        return FileSystem.get(u, conf, StringUtils.isEmpty(hdfsUser) ? "hdfs" : hdfsUser);
    }

    public boolean isConfig(){
        return StringUtils.isNotEmpty(hadoopConfDir);
    }

    /**
     * 是否配置Kerberos认证
     * @return
     */
    public boolean isKerberos(){
        return StringUtils.isNotEmpty(krb5Conf) && StringUtils.isNotEmpty(krb5User) && StringUtils.isNotEmpty(krb5Tab);
    }


    /**
     * 文件列表
     * @param path
     * @return
     * @throws Exception
     */
    public List<String> listFiles(String path) throws Exception {
        List<String> fileNames = new ArrayList<>();
        action(fs -> {
            FileStatus[] fileStatuses = fs.listStatus(new Path(path));
            for (FileStatus fileStatus : fileStatuses) {
                if (fileStatus.isDirectory()) {
                    fileNames.add(fileStatus.getPath().toString() + "/");
                } else {
                    fileNames.add(fileStatus.getPath().toString());
                }
            }
        });
        return fileNames;
    }

    /**
     * hdfs操作
     * @param callback
     * @throws Exception
     */
    public void action(Callback callback) throws Exception {
        try(FileSystem fs = getFileSystem()){
            if(callback != null){
                callback.call(fs);
            }
        }
    }

    @FunctionalInterface
    interface Callback{
        void call(FileSystem fs) throws Exception;
    }
}
